#include "executor.h"
#include <cstring>
#include "utils/like.h"
#include "utils/string_utils.hpp"
#include <algorithm> // 包含std::shuffle
#include <random>    // 包含随机数生成器
#include <chrono>    // 用于获取系统时间作为随机种子
#include <regex>     // 用于日期/时间格式验证
#include <limits>    // 用于 std::numeric_limits
BPlusTreeKey extractKeyFromRow(const Row &row, const IndexSchema &index_schema, const TableSchema &table_schema)
{
    if (index_schema.columns.empty())
    {
        throw std::runtime_error("index " + index_schema.index_name + " has no columns");
    }
    // 简化处理:只取索引的第一个列作为键
    const std::string &col_name = index_schema.columns[0];

    // 找到该列在表结构中的位置
    for (size_t i = 0; i < table_schema.cols.size(); ++i)
    {
        if (table_schema.cols[i].name == col_name)
        {
            return row[i];
        }
    }
    throw std::runtime_error("The index column cannot be found in the row data: " + col_name);
}

// 从一行数据中提取主键
BPlusTreeKey extractPrimaryKeyFromRow(const Row &row, const TableSchema &table_schema)
{
    for (size_t i = 0; i < table_schema.cols.size(); ++i)
    {
        if (table_schema.cols[i].primary)
        {
            return row[i];
        }
    }
    throw std::runtime_error("in table " + table_schema.table_name + " no primary key found");
}

// 辅助函数：尝试将任意 ValueType 转换为 bool
// 返回 std::optional<bool>。如果转换失败，返回 std::nullopt
std::optional<bool> convertToBool(const ValueType &val)
{
    return std::visit([](auto &&arg) -> std::optional<bool>
                      {
                          using T = std::decay_t<decltype(arg)>;

                          if constexpr (std::is_same_v<T, bool>)
                          {
                              return arg; // 本身就是 bool
                          }
                          else if constexpr (std::is_integral_v<T>)
                          {
                              return arg != 0; // 整数：0 为 false，非0 为 true
                          }
                          else if constexpr (std::is_same_v<T, std::string>)
                          {
                              std::string s = toUpperCase(arg); // 转换为大写以便不区分大小写比较
                              if (s == "TRUE" || s == "1")
                              {
                                  return true;
                              }
                              if (s == "FALSE" || s == "0")
                              {
                                  return false;
                              }
                          }

                          // 其他类型 (如 float, double 或无法识别的字符串) 无法安全转换为 bool
                          return std::nullopt; },
                      val);
}

// 辅助函数:验证JSON字符串格式是否正确
bool validateJsonFormat(const std::string &s)
{
    try
    {
        json::parse(s);
        return true;
    }
    catch (const json::parse_error &)
    {
        return false;
    }
}
// 辅助函数：检查是否是闰年
bool is_leap(int year)
{
    return (year % 4 == 0 && year % 100 != 0) || (year % 400 == 0);
}

// 辅助函数:验证日期/时间格式
bool validateDateTimeFormat(const std::string &s, const std::string &format_type)
{
    // --- 步骤 1: 格式检查 ---
    static const std::regex date_regex(R"(^\d{4}-\d{2}-\d{2}$)");
    static const std::regex time_regex(R"(^\d{2}:\d{2}:\d{2}$)");
    static const std::regex datetime_regex(R"(^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$)");

    std::string upper_format_type = toUpperCase(format_type);

    if (upper_format_type == "DATE")
    {
        if (!std::regex_match(s, date_regex))
            return false;
    }
    else if (upper_format_type == "TIME")
    {
        if (!std::regex_match(s, time_regex))
            return false;
    }
    else if (upper_format_type == "DATETIME")
    {
        if (!std::regex_match(s, datetime_regex))
            return false;
    }
    else
    {
        return false; // 不支持的格式类型
    }

    // --- 步骤 2: 语义验证 ---
    int year = 0, month = 0, day = 0, hour = 0, minute = 0, second = 0;

    if (upper_format_type == "DATE" || upper_format_type == "DATETIME")
    {
        sscanf(s.c_str(), "%d-%d-%d", &year, &month, &day);

        if (month < 1 || month > 12)
            return false;
        if (day < 1 || day > 31)
            return false;

        // 检查小月份的天数
        if ((month == 4 || month == 6 || month == 9 || month == 11) && day > 30)
        {
            return false;
        }
        // 检查二月的天数
        if (month == 2)
        {
            if (is_leap(year))
            {
                if (day > 29)
                    return false;
            }
            else
            {
                if (day > 28)
                    return false;
            }
        }
    }

    if (upper_format_type == "TIME" || upper_format_type == "DATETIME")
    {
        // 如果是 DATETIME, sscanf 的格式字符串需要调整
        if (upper_format_type == "DATETIME")
        {
            sscanf(s.c_str(), "%*d-%*d-%*d %d:%d:%d", &hour, &minute, &second);
        }
        else
        { // 是 TIME
            sscanf(s.c_str(), "%d:%d:%d", &hour, &minute, &second);
        }

        if (hour < 0 || hour > 23)
            return false;
        if (minute < 0 || minute > 59)
            return false;
        if (second < 0 || second > 59)
            return false;
    }

    return true;
}

// 辅助函数:将 Value 对象转换为其字符串表示形式
std::string valueToString(const Value &val)
{
    // 如果是 NULL,返回 "NULL" 字符串或一个可识别的标记
    if (!val.has_value())
    {
        // 在类型转换逻辑中,我们只处理非NULL值,但为完整性提供此分支
        throw std::logic_error("valueToString should not be called with a null value");
    }

    // 使用 std::visit 来处理 variant 中的具体类型
    return std::visit([](auto &&arg) -> std::string
                      {
        using T = std::decay_t<decltype(arg)>;
        if constexpr (std::is_same_v<T, std::string>) {
            // 如果本身就是字符串,直接返回
            return arg;
        } else {
            // 对于所有数字类型和bool,使用 std::to_string
            return std::to_string(arg);
        } }, val.value());
}

// 这个函数包含了所有必要的范围检查，是类型转换的核心
Value convertValueToType(const Value &original_value, const std::string &target_sql_type)
{
    if (!original_value.has_value())
    {
        return std::nullopt; // NULL值直接返回，无需转换
    }

    std::string upper_target_type = toUpperCase(target_sql_type);

    // 核心逻辑：使用 visit 访问 variant 内部的具体值
    return std::visit([&](auto &&arg) -> Value
                      {
                          using T = std::decay_t<decltype(arg)>;
                        if (upper_target_type == "BOOL" || upper_target_type == "BOOLEAN") {
                            if constexpr (std::is_same_v<T, bool>) {
                                return ValueType(arg); // 本身就是 bool
                            }
                            if constexpr (std::is_arithmetic_v<T>) {
                                return ValueType(static_cast<long double>(arg) != 0.0L); // 数值：0 为 false, 非0 为 true
                            }
                            if constexpr (std::is_same_v<T, std::string>) {
                                std::string s = toUpperCase(arg);
                                if (s == "TRUE" || s == "1") return ValueType(true);
                                if (s == "FALSE" || s == "0") return ValueType(false);
                                // 如果是其他字符串，则转换失败
                                throw std::invalid_argument("Cannot convert string '" + arg + "' to BOOLEAN.");
                            }
                        }
                          // --- 步骤 1: 将任何输入类型转换为通用的中间表示 ---
                          // 我们使用 long double 作为所有数值的最高精度通用表示
                          long double numeric_representation = 0.0;
                          bool is_convertible_to_numeric = false;

                          if constexpr (std::is_arithmetic_v<T>)
                          {
                              numeric_representation = static_cast<long double>(arg);
                              is_convertible_to_numeric = true;
                          }
                          else if constexpr (std::is_same_v<T, std::string>)
                          {
                              // 尝试将字符串解析为数字
                              const char *start = arg.c_str();
                              const char *end = start + arg.length();
                              std::from_chars_result res = std::from_chars(start, end, numeric_representation);
                              // 如果整个字符串都被成功解析，则认为它是数值
                              if (res.ec == std::errc() && res.ptr == end)
                              {
                                  is_convertible_to_numeric = true;
                              }
                          }

                          // --- 步骤 2: 根据目标类型，从通用表示进行带范围检查的转换 ---

                          // 如果目标是数值类型
                          if (is_convertible_to_numeric)
                          {
                              if (upper_target_type == "TINYINT")
                              {
                                  if (numeric_representation < std::numeric_limits<int8_t>::min() || numeric_representation > std::numeric_limits<int8_t>::max())
                                      throw std::out_of_range("Value is out of range for TINYINT.");
                                  return ValueType(static_cast<int8_t>(numeric_representation));
                              }
                              if (upper_target_type == "TINYINT UNSIGNED")
                              {
                                  if (numeric_representation < 0 || numeric_representation > std::numeric_limits<uint8_t>::max())
                                      throw std::out_of_range("Value is out of range for TINYINT UNSIGNED.");
                                  return ValueType(static_cast<uint8_t>(numeric_representation));
                              }
                              if (upper_target_type == "SMALLINT")
                              {
                                  if (numeric_representation < std::numeric_limits<int16_t>::min() || numeric_representation > std::numeric_limits<int16_t>::max())
                                      throw std::out_of_range("Value is out of range for SMALLINT.");
                                  return ValueType(static_cast<int16_t>(numeric_representation));
                              }
                              if (upper_target_type == "SMALLINT UNSIGNED")
                              {
                                  if (numeric_representation < 0 || numeric_representation > std::numeric_limits<uint16_t>::max())
                                      throw std::out_of_range("Value is out of range for SMALLINT UNSIGNED.");
                                  return ValueType(static_cast<uint16_t>(numeric_representation));
                              }
                              if (upper_target_type == "INT")
                              {
                                  if (numeric_representation < std::numeric_limits<int32_t>::min() || numeric_representation > std::numeric_limits<int32_t>::max())
                                      throw std::out_of_range("Value is out of range for INT.");
                                  return ValueType(static_cast<int32_t>(numeric_representation));
                              }
                              if (upper_target_type == "INT UNSIGNED")
                              {
                                  if (numeric_representation < 0 || numeric_representation > std::numeric_limits<uint32_t>::max())
                                      throw std::out_of_range("Value is out of range for INT UNSIGNED.");
                                  return ValueType(static_cast<uint32_t>(numeric_representation));
                              }
                              if (upper_target_type == "BIGINT")
                              {
                                  if (numeric_representation < std::numeric_limits<int64_t>::min() || numeric_representation > std::numeric_limits<int64_t>::max())
                                      throw std::out_of_range("Value is out of range for BIGINT.");
                                  return ValueType(static_cast<int64_t>(numeric_representation));
                              }
                              if (upper_target_type == "BIGINT UNSIGNED")
                              {
                                  if (numeric_representation < 0 || numeric_representation > std::numeric_limits<uint64_t>::max())
                                      throw std::out_of_range("Value is out of range for BIGINT UNSIGNED.");
                                  return ValueType(static_cast<uint64_t>(numeric_representation));
                              }
                              if (upper_target_type == "FLOAT")
                              {
                                  if (numeric_representation < -std::numeric_limits<float>::max() || numeric_representation > std::numeric_limits<float>::max())
                                      throw std::out_of_range("Value is out of range for FLOAT.");
                                  return ValueType(static_cast<float>(numeric_representation));
                              }
                              if (upper_target_type == "DOUBLE")
                              {
                                  if (numeric_representation < -std::numeric_limits<double>::max() || numeric_representation > std::numeric_limits<double>::max())
                                      throw std::out_of_range("Value is out of range for DOUBLE.");
                                  return ValueType(static_cast<double>(numeric_representation));
                              }
                          }

                          // 如果目标是字符串类型
                          if (upper_target_type.find("CHAR") != std::string::npos ||
                              upper_target_type.find("TEXT") != std::string::npos ||
                              upper_target_type == "JSON" || // JSON is stored as a string
                              upper_target_type == "DATE" ||
                              upper_target_type == "TIME" ||
                              upper_target_type == "DATETIME")
                          {
                              // 如果原始类型是字符串，直接返回（后续可以在这里加长度检查）
                              if constexpr (std::is_same_v<T, std::string>)
                              {
                                  return ValueType(arg);
                              }
                              // 如果原始类型是数值，转换为字符串
                              if constexpr (std::is_arithmetic_v<T>)
                              {
                                  return ValueType(std::to_string(arg));
                              }
                          }

                          // 如果以上所有转换都失败，则抛出异常
                          throw std::invalid_argument("Unsupported or invalid type conversion from original type to '" + target_sql_type + "'."); },
                      original_value.value());
}

Value parseValueFromString(const std::string &str_val, const std::string &sql_type)
{
    Value temp_val = ValueType(str_val);
    return convertValueToType(temp_val, sql_type);
}

QueryResult Executor::executePlan(const ExecutionPlanNode &plan)
{
    try
    {
        return executeOp(plan);
    }
    catch (const std::exception &e)
    {
        logger_.Error("Execute plan failed: " + std::string(e.what()));
        QueryResult err;
        err.status = "error";
        err.error_type = "Execution Error";
        err.message = e.what();
        return err;
    }
}

QueryResult Executor::executeOp(const ExecutionPlanNode &plan)
{
    QueryResult result;
    switch (plan.op_type)
    {
    case OpType::CreateTable:
    {
        result.type = QueryType::CREATE_TABLE;
        try
        {
            bool ok = executeCreateTable(
                plan.details.table_name,
                plan.details.create_cols,
                plan.details.create_comment);
            if (ok)
            {
                result.status = "success";
                result.message = "Table created successfully.";
            }
            else
            {
                result.status = "error";
                result.error_type = "Create Table Error";
                result.message = "Failed to create table.";
            }
        }
        catch (const std::exception &e)
        {
            result.status = "error";
            result.error_type = "Create Table Error";
            result.message = e.what();
        }
        break;
    }
    case OpType::Insert:
    {
        result.type = QueryType::INSERT;
        try
        {
            bool ok = executeInsert(
                plan.details.table_name,
                plan.details.insert_values);
            if (ok)
            {
                result.status = "success";
                result.message = "1 row affected.";
            }
            else
            {
                result.status = "error";
                result.error_type = "Insert Error";
                result.message = "Failed to insert row.";
            }
        }
        catch (const std::exception &e)
        {
            result.status = "error";
            result.error_type = "Insert Error";
            result.message = e.what();
        }
        break;
    }
    case OpType::Delete:
    {
        result.type = QueryType::DELETE;
        try
        {
            size_t rows_affected = executeDelete(
                plan.details.table_name,
                plan.details.condition);
            if (rows_affected >= 0)
            {
                result.status = "success";
                result.message = std::to_string(rows_affected) + " row(s) affected.";
            }
            else
            {
                result.status = "error";
                result.error_type = "Delete Error";
                result.message = "Delete row failed.";
            }
        }
        catch (const std::exception &e)
        {
            result.status = "error";
            result.error_type = "Delete Error";
            result.message = e.what();
        }
        break;
    }
    case OpType::Update:
    {
        result.type = QueryType::UPDATE;
        try
        {
            size_t rows_affected = executeUpdate(
                plan.details.table_name,
                plan.details.condition,
                plan.details.update_values);
            if (rows_affected >= 0)
            {
                result.status = "success";
                result.message = std::to_string(rows_affected) + " row(s) affected.";
            }
            else
            {
                result.status = "error";
                result.error_type = "Update Error";
                result.message = "Update row failed.";
            }
        }
        catch (const std::exception &e)
        {
            result.status = "error";
            result.error_type = "Update Error";
            result.message = e.what();
        }
        break;
    }
    case OpType::Project:
    {
        try
        {
            // 递归执行子算子(如Filter)
            auto child_result = executeOp(*plan.child);
            if (child_result.status != "success")
                return child_result;
            // 执行Project算子
            auto project_rows = executeProject(
                child_result.rows,
                plan.details.project_cols,
                plan.child->details.table_name,
                plan.details.offset,
                plan.details.limit,
                plan.details.order_by);
            result.status = "success";
            if (plan.details.project_cols.size() == 1 && plan.details.project_cols[0] == "*")
            {
                std::vector<std::string> col_names;
                auto cols = catalog_.getTableSchema(plan.child->details.table_name).cols;
                for (auto &col : cols)
                {
                    col_names.push_back(col.name);
                }
                result.columns = col_names;
            }
            else
            {
                result.columns = plan.details.project_cols;
            }
            result.rows = project_rows;
            result.message = "Success get rows:" + std::to_string(project_rows.size());
        }
        catch (const std::exception &e)
        {
            result.status = "error";
            result.error_type = "Project Error";
            result.message = e.what();
        }
        break;
    }
    case OpType::Filter:
    {
        try
        {
            // 执行Filter算子
            auto filter_rows = executeFilter(
                plan.details.condition,
                plan.details.table_name);
            result.status = "success";
            // 获取表的列名
            auto table_schema = catalog_.getTableSchema(plan.details.table_name);
            for (const auto &col : table_schema.cols)
            {
                result.columns.push_back(col.name);
            }
            result.rows = filter_rows;
            result.message = "Success get rows:" + std::to_string(filter_rows.size());
        }
        catch (const std::exception &e)
        {
            result.status = "error";
            result.error_type = "Filter Error";
            result.message = e.what();
        }
        break;
    }
    case OpType::SeqScan:
    {
        try
        {
            auto rows = executeSeqScan(plan.details.table_name);
            result.status = "success";
            // 获取表的列名
            auto table_schema = catalog_.getTableSchema(plan.details.table_name);
            for (const auto &col : table_schema.cols)
            {
                result.columns.push_back(col.name);
            }
            result.rows = rows;
            result.message = "Success get rows:" + std::to_string(rows.size());
        }
        catch (const std::exception &e)
        {
            result.status = "error";
            result.error_type = "SeqScan Error";
            result.message = e.what();
        }

        break;
    }
    case OpType::CreateIndex:
    {
        result.type = QueryType::CREATE_INDEX;
        try
        {
            bool ok = executeCreateIndex(
                plan.details.table_name,
                plan.details.index_name,
                plan.details.index_columns,
                plan.details.is_unique);
            if (ok)
            {
                result.status = "success";
                result.message = "Index created successfully.";
            }
            else
            {
                result.status = "error";
                result.error_type = "Create Index Error";
                result.message = "Failed to create index.";
            }
        }
        catch (const std::exception &e)
        {
            result.status = "error";
            result.error_type = "Create Index Error";
            result.message = e.what();
        }

        break;
    }

    case OpType::DropIndex:
    {
        result.type = QueryType::DROP_INDEX;
        try
        {
            bool ok = executeDropIndex(
                plan.details.table_name,
                plan.details.index_name);
            if (ok)
            {
                result.status = "success";
                result.message = "Index dropped successfully.";
            }
            else
            {
                result.status = "error";
                result.error_type = "Drop Index Error";
                result.message = "Failed to drop index.";
            }
        }
        catch (const std::exception &e)
        {
            result.status = "error";
            result.error_type = "Drop Index Error";
            result.message = e.what();
        }
        break;
    }
    case OpType::DropTable:
    {
        result.type = QueryType::DROP_TABLE;
        try
        {
            bool ok = executeDropTable(
                plan.details.table_name);
            if (ok)
            {
                result.status = "success";
                result.message = "Table dropped successfully.";
            }
            else
            {
                result.status = "error";
                result.error_type = "Drop Table Error";
                result.message = "Failed to drop table.";
            }
        }
        catch (const std::exception &e)
        {
            result.status = "error";
            result.error_type = "Drop Table Error";
            result.message = e.what();
        }
        break;
    }
    case OpType::ShowTables:
    {
        result.type = QueryType::SHOW_TABLES;
        try
        {
            result.status = "success";
            result.schemas = executeShowTables();
            result.message = "Success get tables.number:" + std::to_string(result.schemas.size());
        }
        catch (const std::exception &e)
        {
            result.status = "error";
            result.error_type = "Show Tables Error";
            result.message = e.what();
        }
        break;
    }
    case OpType::ShowIndices:
    {
        result.type = QueryType::SHOW_INDICES;
        try
        {
            result.status = "success";
            result.indexes = executeShowIndexes(plan.details.table_name);
            result.message = "Success get indexes.number:" + std::to_string(result.indexes.size());
        }
        catch (const std::exception &e)
        {
            result.status = "error";
            result.error_type = "Show Indexes Error";
            result.message = e.what();
        }
        break;
    }
    case OpType::AlterTable:
    {
        result.type = QueryType::ALTER;
        try
        {
            bool ok = executeAlterTable(
                plan.details.table_name,
                plan.details.rename_table_name,
                plan.details.modified_comment,
                plan.details.modified_cols,
                plan.details.modified_indexes,
                plan.details.drop_cols,
                plan.details.add_cols);
            if (ok)
            {
                result.status = "success";
                result.message = "Alter table successfully.";
            }
            else
            {
                result.status = "error";
                result.error_type = "Alter Table Error";
                result.message = "Failed to alter table.";
            }
        }
        catch (const std::exception &e)
        {
            result.status = "error";
            result.error_type = "Alter Table Error";
            result.message = e.what();
        }
        break;
    }
    default:
        throw std::runtime_error("Unsupported operator type.");
    }
    return result;
}

bool Executor::executeCreateTable(const std::string &table_name, const std::vector<ColumnSchema> &cols, const std::string &comment)
{
    // 1. 判断表是否存在
    if (catalog_.tableExists(table_name))
    {
        logger_.Error("Table '" + table_name + "' already exists.");
        throw std::runtime_error("Table '" + table_name + "' already exists.");
    }
    // 2. 判断列定义合法性
    if (cols.empty())
    {
        logger_.Error("Table '" + table_name + "' has no columns.");
        throw std::runtime_error("Table '" + table_name + "' has no columns.");
    }
    int primary_key_count = 0;
    for (auto &col : cols)
    {
        if (col.primary)
        {
            primary_key_count++;
        }
        validateColumn(col);
    }
    if (primary_key_count > 1)
    {
        logger_.Error("Table '" + table_name + "' has multiple primary keys.");
        throw std::runtime_error("Table '" + table_name + "' has multiple primary keys.");
    }
    return catalog_.createTable(table_name, cols, comment);
}

bool Executor::executeInsert(const std::string &table_name, const std::unordered_map<std::string, Value> &values)
{
    // 1. 检查表是否存在和插入值类型合法性
    auto table_schema = catalog_.getTableSchema(table_name);

    Row row;
    auto check_result = checkInsertType(values, table_name, row);
    if (!check_result.first)
    {
        throw std::runtime_error(check_result.second);
    }

    // 2. 提取主键
    BPlusTreeKey primary_key = extractPrimaryKeyFromRow(row, table_schema);

    // 3. 插入到主键索引
    BPlusTree primary_tree(table_name, table_schema.primary_key_name, catalog_);
    primary_tree.Insert(primary_key, row, catalog_);
    // 4. 遍历所有二级索引,并插入相应的条目
    for (const auto &index_schema : table_schema.indexes)
    {
        // 跳过主键索引,因为它已经处理过了
        if (index_schema.index_name == table_schema.primary_key_name)
        {
            continue;
        }

        // 提取二级索引键
        BPlusTreeKey secondary_key = extractKeyFromRow(row, index_schema, table_schema);

        // 插入到二级索引树
        BPlusTree secondary_tree(table_name, index_schema.index_name, catalog_);
        secondary_tree.Insert(secondary_key, primary_key, catalog_);
    }

    return true;
}

std::vector<std::vector<Value>> Executor::executeSeqScan(const std::string &table_name)
{
    // 1. 获取表结构
    auto table_schema = catalog_.getTableSchema(table_name);
    // 2. 构建B+树
    BPlusTree bptree(table_name, table_schema.primary_key_name, catalog_);
    // 3. 扫描所有数据
    auto all_rows = bptree.ScanAll();
    return std::get<std::vector<Row>>(all_rows);
}
uint32_t getColumnIndex(const TableSchema &schema, const std::string &column_name)
{
    size_t size = schema.cols.size();
    for (size_t i = 0; i < size; i++)
    {
        if (schema.cols[i].name == column_name)
        {
            return i;
        }
    }
    throw std::runtime_error("Column '" + column_name + "' not found.");
}
bool Executor::evaluateCondition(const Row &row, const TableSchema &schema, const Condition &cond)
{
    // 递归的终止条件
    if (cond.type == ConditionType::SIMPLE)
    {
        // 1. 处理 IS NULL / IS NOT NULL
        if (toUpperCase(cond.op) == "IS NULL")
        {
            size_t col_index = getColumnIndex(schema, cond.column);
            return !row[col_index].has_value(); // 值为 std::nullopt 即为 NULL
        }
        if (toUpperCase(cond.op) == "IS NOT NULL")
        {
            size_t col_index = getColumnIndex(schema, cond.column);
            return row[col_index].has_value(); // 值存在即为 NOT NULL
        }

        // 2. 获取列的当前值
        size_t col_index = getColumnIndex(schema, cond.column);
        const Value &left_val = row[col_index];
        const Value &right_val = cond.value;

        // 3. SQL标准:任何与 NULL (std::nullopt) 的常规比较都为 false
        if (!left_val.has_value() || !right_val.has_value())
        {
            return false;
        }

        // 4. 访问 variant 内部,进行类型感知的比较
        return std::visit([&cond](auto &&left, auto &&right) -> bool
                          {
                              using LeftT = std::decay_t<decltype(left)>;
                              using RightT = std::decay_t<decltype(right)>;

                              // --- 分支 1: 布尔逻辑比较 ---
                              // 只要有一方是 bool，就尝试进行布尔比较
                              if constexpr (std::is_same_v<LeftT, bool> || std::is_same_v<RightT, bool>)
                              {
                                  std::optional<bool> left_b = convertToBool(left);
                                  std::optional<bool> right_b = convertToBool(right);

                                  // 如果任意一方无法转换为 bool，则比较失败
                                  if (!left_b.has_value() || !right_b.has_value())
                                  {
                                      return false;
                                  }

                                  if (cond.op == "=")
                                      return left_b.value() == right_b.value();
                                  if (cond.op == "!=" || cond.op == "<>")
                                      return left_b.value() != right_b.value();

                                  // > < >= <= 对布尔类型无意义，返回 false
                                  return false;
                              }

                              // --- 分支 2: 字符串 'LIKE' 比较 ---
                              if (toUpperCase(cond.op) == "LIKE")
                              {
                                  if constexpr (std::is_same_v<LeftT, std::string> && std::is_same_v<RightT, std::string>)
                                      return isMysqlLike(left, right);
                                  else
                                      return false; // LIKE 只适用于字符串
                              }

                              // --- 分支 3: 纯数字 (算术) 比较 ---
                              // (排除了布尔和字符串比较后)
                              if constexpr (std::is_arithmetic_v<LeftT> && std::is_arithmetic_v<RightT>)
                              {
                                  // 为了兼容所有整数和浮点数，统一提升到 double 进行比较
                                  // 注意：对于超大整数 (int64_t)，这可能损失精度，但对于大多数场景是可行的
                                  double left_d = static_cast<double>(left);
                                  double right_d = static_cast<double>(right);
                                  if (cond.op == "=")
                                      return left_d == right_d;
                                  if (cond.op == "!=" || cond.op == "<>")
                                      return left_d != right_d;
                                  if (cond.op == ">")
                                      return left_d > right_d;
                                  if (cond.op == "<")
                                      return left_d < right_d;
                                  if (cond.op == ">=")
                                      return left_d >= right_d;
                                  if (cond.op == "<=")
                                      return left_d <= right_d;
                              }

                              // --- 分支 4: 字符串的通用比较 ---
                              // 如果以上都不是，最后尝试将两边都转为字符串进行字典序比较
                              // 这可以处理 DATE/TIME 等以字符串形式存储的类型
                              if constexpr (std::is_same_v<LeftT, std::string> || std::is_same_v<RightT, std::string>)
                              {
                                  std::stringstream ss_left, ss_right;
                                  ss_left << left;
                                  ss_right << right;
                                  std::string s_left = ss_left.str();
                                  std::string s_right = ss_right.str();

                                  if (cond.op == "=")
                                      return s_left == s_right;
                                  if (cond.op == "!=" || cond.op == "<>")
                                      return s_left != s_right;
                                  if (cond.op == ">")
                                      return s_left > s_right;
                                  if (cond.op == "<")
                                      return s_left < s_right;
                                  if (cond.op == ">=")
                                      return s_left >= s_right;
                                  if (cond.op == "<=")
                                      return s_left <= s_right;
                              }

                              // 所有情况都无法处理，则类型不兼容
                              return false; },
                          left_val.value(), right_val.value());
    }

    if (cond.type == ConditionType::AND)
    {
        for (const auto &child_cond : cond.children)
        {
            if (!evaluateCondition(row, schema, *child_cond))
            {
                return false;
            }
        }
        return true;
    }

    if (cond.type == ConditionType::OR)
    {
        for (const auto &child_cond : cond.children)
        {
            if (evaluateCondition(row, schema, *child_cond))
            {
                return true;
            }
        }
        return false;
    }
    return false;
}
/**
 * @brief 递归遍历条件树,寻找一个最适合使用索引的简单条件.
 * @param cond 条件树的当前节点.
 * @param schema 表结构,用于检查列是否存在索引.
 * @return 返回找到的最佳简单条件的指针,如果没找到则返回 nullptr.
 *
 * 优化策略:
 * 1. 优先选择 `=` 操作的条件.
 * 2. 其次选择范围操作('>', '<', '>=', '<=')的条件.
 * 3. 只在 AND 条件链中寻找,OR 条件不进行索引优化.
 */
const Condition *Executor::findIndexableCondition(const Condition &cond, const TableSchema &schema)
{
    if (cond.type == ConditionType::SIMPLE)
    {
        // 检查该列是否有索引
        for (const auto &index : schema.indexes)
        {
            // 简化:只考虑单列索引
            if (index.columns.size() == 1 && index.columns[0] == cond.column && cond.op != "!=" && cond.op != "<>")
            {
                return &cond; // 找到了一个可以用索引的简单条件
            }
        }
        return nullptr; // 该列没有索引
    }

    if (cond.type == ConditionType::AND)
    {
        const Condition *best_cond = nullptr;
        // 在 AND 的所有子条件中寻找最优的一个
        for (const auto &child : cond.children)
        {
            const Condition *found_cond = findIndexableCondition(*child, schema);
            if (found_cond)
            {
                // 简单的优化:'=' 比范围查询更好.如果已经找到了 '=',就不再更换.
                if (best_cond == nullptr)
                {
                    best_cond = found_cond;
                }
                else if (best_cond->op != "=" && found_cond->op == "=")
                {
                    best_cond = found_cond; // 发现一个更优的 '=' 条件
                }
            }
        }
        return best_cond;
    }

    // 对于 OR 条件,选择不使用索引,直接返回 nullptr
    return nullptr;
}
std::vector<std::vector<Value>> Executor::executeFilter(
    const Condition &cond,
    const std::string &table_name)
{
    auto table_schema = catalog_.getTableSchema(table_name);
    BPlusTree primary_tree(table_name, table_schema.primary_key_name, catalog_);

    std::vector<Row> candidate_rows;
    bool index_used = false;
    // ==========================================================
    // 阶段一:索引搜索 - 尝试用最优索引缩小候选集范围
    // ==========================================================
    const Condition *best_cond_for_index = findIndexableCondition(cond, table_schema);

    if (best_cond_for_index)
    {
        const auto &indexable_cond = *best_cond_for_index;
        auto index_schema = catalog_.getIndexSchemaByColumn(table_name, indexable_cond.column);

        logger_.Info("查询优化: 在列 '" + indexable_cond.column + "' 上使用索引 '" + index_schema.index_name + "'");
        index_used = true;

        if (index_schema.index_name == table_schema.primary_key_name)
        {
            // ** 使用主键索引 **
            if (indexable_cond.op == "=")
            {
                Row row;
                if (primary_tree.Search(indexable_cond.value, row))
                {
                    candidate_rows.push_back(row);
                }
            }
            else
            { // 范围查询
                // ... 实现主键索引的范围查询 ...
                BPlusTreeKey start_key, end_key;
                bool left_inclusive = false, right_inclusive = false;
                const auto &value = best_cond_for_index->value;

                // 根据操作符和值的类型设置范围
                if (best_cond_for_index->op == ">")
                {
                    start_key = value;
                    left_inclusive = false;
                }
                else if (best_cond_for_index->op == ">=")
                {
                    start_key = value;
                    left_inclusive = true;
                }
                else if (best_cond_for_index->op == "<")
                {
                    end_key = value;
                    right_inclusive = false;
                }
                else if (best_cond_for_index->op == "<=")
                {
                    end_key = value;
                    right_inclusive = true;
                }
                else
                {
                    throw std::runtime_error("unknown operator: " + indexable_cond.op);
                }
                auto res = primary_tree.SearchRange(start_key, end_key, left_inclusive, right_inclusive);
                candidate_rows = std::get<std::vector<Row>>(res);
            }
        }
        else
        {
            // ** 使用二级索引 **
            BPlusTree secondary_tree(table_name, index_schema.index_name, catalog_);
            std::vector<BPlusTreeKey> primary_keys;

            if (indexable_cond.op == "=")
            {
                secondary_tree.Search(indexable_cond.value, primary_keys);
            }
            else
            { // 范围查询
                BPlusTreeKey start_key, end_key;
                bool left_inclusive = false, right_inclusive = false;
                const auto &value = best_cond_for_index->value;

                // 根据操作符和值的类型设置范围
                if (best_cond_for_index->op == ">")
                {
                    start_key = value;
                    left_inclusive = false;
                }
                else if (best_cond_for_index->op == ">=")
                {
                    start_key = value;
                    left_inclusive = true;
                }
                else if (best_cond_for_index->op == "<")
                {
                    end_key = value;
                    right_inclusive = false;
                }
                else if (best_cond_for_index->op == "<=")
                {
                    end_key = value;
                    right_inclusive = true;
                }
                else
                {
                    throw std::runtime_error("unknown operator: " + indexable_cond.op);
                }
                auto res = secondary_tree.SearchRange(start_key, end_key, left_inclusive, right_inclusive);
                primary_keys = std::get<std::vector<BPlusTreeKey>>(res);
            }

            // "回表" 操作: 用找到的主键去主键索引中查完整数据
            for (const auto &pk : primary_keys)
            {
                Row row;
                if (primary_tree.Search(pk, row))
                {
                    candidate_rows.push_back(row);
                }
            }
        }
    }
    if (index_used && cond.type == ConditionType::SIMPLE)
    {
        return candidate_rows;
    }
    // 如果没有找到可用的索引,则退化为全表扫描
    if (!index_used)
    {
        logger_.Info("在表" + table_name + "上的查询未使用索引,将执行全表扫描.");
        auto res = primary_tree.ScanAll();
        candidate_rows = std::get<std::vector<Row>>(res);
    }

    // ==========================================================
    // 阶段二:内存过滤 - 对候选集应用完整的条件
    // ==========================================================
    std::vector<Row> final_rows;
    if (candidate_rows.empty())
    {
        return final_rows; // 候选集为空,直接返回
    }

    for (const auto &row : candidate_rows)
    {
        // 使用完整的、原始的条件树 `cond` 进行最终校验
        if (evaluateCondition(row, table_schema, cond))
        {
            final_rows.push_back(row);
        }
    }

    return final_rows;
}

std::vector<std::vector<Value>> Executor::executeProject(
    std::vector<std::vector<Value>> &rows,
    const std::vector<std::string> &project_cols,
    const std::string &table_name,
    const std::optional<uint16_t> &offset, const std::optional<uint16_t> &limit,
    const std::vector<OrderBy> &orders)
{
    bool is_random_sort = false;

    // --- 1. 处理排序逻辑 ---

    // 优先检查是否存在 RAND 排序
    for (const auto &order : orders)
    {
        if (order.order_type.has_value() && toUpperCase(order.order_type.value()) == "RAND")
        {
            uint16_t sd = order.seed.value_or(
                std::chrono::system_clock::now().time_since_epoch().count());
            std::default_random_engine rng(sd);
            std::shuffle(rows.begin(), rows.end(), rng);

            is_random_sort = true;
            break; // 找到 RAND,不再处理其他排序规则
        }
    }
    auto table_schema = catalog_.getTableSchema(table_name);

    // 如果不是随机排序,并且有排序规则,则执行标准排序
    if (!is_random_sort && !orders.empty())
    {
        // 获取表结构,建立列名到索引的映射以提高效率
        std::map<std::string, int> col_name_to_index;
        for (size_t i = 0; i < table_schema.cols.size(); ++i)
        {
            col_name_to_index[table_schema.cols[i].name] = i;
        }

        // 预处理排序规则:将列名转换为索引,并确定排序方向
        // std::pair<column_index, is_descending>
        std::vector<std::pair<int, bool>> sort_criteria;
        for (const auto &order : orders)
        {
            if (!order.order_by.has_value())
                continue; // 跳过无效的排序规则

            auto it = col_name_to_index.find(order.order_by.value());
            if (it == col_name_to_index.end())
            {
                throw std::runtime_error("order by column '" + order.order_by.value() + "' is not existed in table " + table_name);
            }
            int index = it->second;

            bool is_desc = false;
            if (order.order_type.has_value())
            {
                std::string type = toUpperCase(order.order_type.value());
                if (type == "DESC")
                {
                    is_desc = true;
                }
                else if (type != "ASC")
                {
                    throw std::runtime_error("the type for order by is not valid: " + type);
                }
            }
            sort_criteria.push_back({index, is_desc});
        }

        // 执行多级排序
        if (!sort_criteria.empty())
        {
            std::sort(rows.begin(), rows.end(), [&](const std::vector<Value> &a, const std::vector<Value> &b)
                      {
                          for (const auto &criterion : sort_criteria)
                          {
                              int index = criterion.first;
                              bool is_desc = criterion.second;

                              if (a[index] < b[index])
                              {
                                  return !is_desc; // 如果是 ASC,a < b 返回 true
                              }
                              if (b[index] < a[index])
                              {
                                  return is_desc; // 如果是 DESC,b < a 返回 true
                              }
                              // 如果当前列的值相等,则继续比较下一个排序列
                          }
                          return false; // 所有排序列都相等
                      });
        }
    }

    // --- 2. 处理 LIMIT 和 OFFSET ---
    std::vector<std::vector<Value>> result_rows;
    size_t start_index = offset.value_or(0);

    if (start_index >= rows.size())
    {
        return {};
    }

    size_t end_index = rows.size();
    if (limit.has_value())
    {
        end_index = std::min(rows.size(), start_index + limit.value());
    }

    // 创建一个子视图或复制数据
    if (start_index < end_index)
    {
        result_rows.assign(rows.begin() + start_index, rows.begin() + end_index);
    }
    else
    {
        return {};
    }
    if (project_cols.size() == 1 && project_cols[0] == "*")
    {
        return result_rows;
    }
    std::vector<std::vector<Value>> projected_rows;
    std::vector<size_t> col_indices;
    for (const auto &col_name : project_cols)
    {
        bool found = false;
        for (size_t i = 0; i < table_schema.cols.size(); ++i)
        {
            if (table_schema.cols[i].name == col_name)
            {
                col_indices.push_back(i);
                found = true;
                break;
            }
        }
        if (!found)
        {
            logger_.Error("Column " + col_name + " not found in table " + table_name);
            return {};
        }
    }

    // 2. 对每行提取目标列
    for (const auto &row : result_rows)
    {
        std::vector<Value> projected_row;
        for (size_t idx : col_indices)
        {
            if (idx < row.size())
            {
                projected_row.push_back(row[idx]);
            }
            else
            {
                logger_.Error("Row has insufficient columns for projection.");
                projected_row.clear();
                break;
            }
        }
        if (!projected_row.empty())
        {
            projected_rows.push_back(projected_row);
        }
    }

    return projected_rows;
}

// 辅助函数,用于根据列定义获取 NULL 值
Value getNullValueForType(const ColumnSchema &col)
{
    if (col.nullable)
    {
        return std::nullopt;
    }
    // 这是一个错误情况,非空列必须有值
    throw std::runtime_error("Column " + col.name + " is not nullable, but no value was provided.");
}
std::pair<bool, std::string> Executor::checkInsertType(const std::unordered_map<std::string, Value> &values, const std::string &table_name, std::vector<Value> &row)
{
    try
    {
        // 从系统目录获取表结构
        auto table_schema = catalog_.getTableSchema(table_name);
        row.clear();
        // 初始化row,大小与列定义一致,所有值默认为 "尚未提供"
        row.resize(table_schema.cols.size(), std::nullopt);

        // --- 阶段 1: 将用户提供的值填入 row 的正确位置 ---
        for (const auto &[col_name, val] : values)
        {
            size_t index = getColumnIndex(table_schema, col_name);
            row[index] = val;
        }

        // --- 阶段 2: 遍历所有列定义,处理缺失值、检查约束和验证类型 ---
        for (size_t i = 0; i < table_schema.cols.size(); ++i)
        {
            const auto &col = table_schema.cols[i];
            Value &current_val = row[i];

            // 检查值是否由用户提供
            bool value_provided = values.count(col.name);
            // 如果值未提供,应用默认规则
            if (!value_provided)
            {
                if (col.auto_increment)
                {
                    int64_t count = catalog_.getAutoIncrementCounter(table_name, col.name);
                    if (col.type == "TINYINT")
                    {
                        current_val = ValueType(static_cast<int8_t>(count));
                    }
                    else if (col.type == "SMALLINT")
                    {
                        current_val = ValueType(static_cast<int16_t>(count));
                    }
                    else if (col.type == "INT")
                    {
                        current_val = ValueType(static_cast<int32_t>(count));
                    }
                    else if (col.type == "BIGINT")
                    {
                        current_val = ValueType(count);
                    }
                    else if (col.type == "TINYINT UNSIGNED")
                    {
                        current_val = ValueType(static_cast<uint8_t>(count));
                    }
                    else if (col.type == "SMALLINT UNSIGNED")
                    {
                        current_val = ValueType(static_cast<uint16_t>(count));
                    }
                    else if (col.type == "INT UNSIGNED")
                    {
                        current_val = ValueType(static_cast<uint32_t>(count));
                    }
                    else if (col.type == "BIGINT UNSIGNED")
                    {
                        current_val = ValueType(static_cast<uint64_t>(count));
                    }
                    else
                    {
                        throw std::runtime_error("Unsupported auto_increment type: " + col.type);
                    }
                }
                else if (col.default_value.has_value())
                {
                    // 使用默认值
                    current_val = parseValueFromString(col.default_value.value(), col.type);
                }
                else if (col.nullable)
                {
                    // 如果可空,则值为 NULL
                    current_val = std::nullopt;
                }
                else
                {
                    // 其他情况(非空,无默认值,非自增),这是错误
                    return {false, "Column '" + col.name + "' is not nullable, and no value was provided."};
                }
            }
            std::string upper_col_type = toUpperCase(col.type);
            // *** 特别处理 BOOLEAN 类型的类型转换 ***
            if (upper_col_type == "BOOLEAN" || upper_col_type == "BOOL")
            {
                if (current_val.has_value())
                {
                    std::optional<bool> converted_bool = convertToBool(current_val.value());
                    if (converted_bool.has_value())
                    {
                        // 转换成功，用真正的 bool 值替换原来的值
                        current_val = ValueType(converted_bool.value());
                    }
                    else
                    {
                        // 转换失败，说明输入的值对于布尔列是无效的
                        std::string original_value_str = valueToString(current_val); // 获取原始值的字符串表示
                        return {false, "列 '" + col.name + "' 的值 '" + original_value_str + "' 不是一个有效的布尔值。"};
                    }
                }
            }
            // 检查 NOT NULL 约束
            if (!col.nullable && !current_val.has_value())
            {
                return {false, "Column '" + col.name + "' violated the NOT NULL constraint."};
            }

            // 如果值存在 (非NULL), 进行类型和格式的详细验证
            if (current_val.has_value())
            {

                std::visit([&](auto &&arg)
                           {
                               using T = std::decay_t<decltype(arg)>;

                               // 整数范围检查
                               if constexpr (std::is_integral_v<T> && !std::is_same_v<T, bool>)
                               {
                                   int64_t v = static_cast<int64_t>(arg);
                                   if (upper_col_type == "TINYINT" && (v < std::numeric_limits<int8_t>::min() || v > std::numeric_limits<int8_t>::max()))
                                       throw std::runtime_error("value out of TINYINT range for column '" + col.name + "'");
                                   if (upper_col_type == "SMALLINT" && (v < std::numeric_limits<int16_t>::min() || v > std::numeric_limits<int16_t>::max()))
                                       throw std::runtime_error("value out of SMALLINT range for column '" + col.name + "'");
                                   if(upper_col_type == "INT"&& (v < std::numeric_limits<int32_t>::min() || v > std::numeric_limits<int32_t>::max()))
                                        throw std::runtime_error("value out of INT range for column '" + col.name + "'");
                                   if(upper_col_type == "BIGINT"&& (v < std::numeric_limits<int64_t>::min() || v > std::numeric_limits<int64_t>::max()))
                                       throw std::runtime_error("value out of BIGINT range for column '" + col.name + "'");
                                   if(upper_col_type =="TINYINT UNSIGNED"&& (v < 0 || v > std::numeric_limits<uint8_t>::max()))
                                       throw std::runtime_error("value out of TINYINT UNSIGNED range for column '" + col.name + "'");
                                    if(upper_col_type =="SMALLINT UNSIGNED"&& (v < 0 || v > std::numeric_limits<uint16_t>::max()))
                                    throw std::runtime_error("value out of SMALLINT UNSIGNED range for column '" + col.name + "'");
                                    if(upper_col_type =="INT UNSIGNED"&& (v < 0 || v > std::numeric_limits<uint32_t>::max()))
                                    throw std::runtime_error("value out of INT UNSIGNED range for column '" + col.name + "'");
                                    if(upper_col_type =="BIGINT UNSIGNED"&& (v < 0 || v > std::numeric_limits<uint64_t>::max()))
                                    throw std::runtime_error("value out of BIGINT UNSIGNED range for column '" + col.name + "'");
                               }

                               // 字符串相关检查
                               if constexpr (std::is_same_v<T, std::string>)
                               {
                                   if ((upper_col_type == "VARCHAR"|| upper_col_type == "CHAR") && col.length > 0 && arg.length() > col.length)
                                   {
                                       throw std::runtime_error("sting length(" + std::to_string(arg.length()) + ")  exceeds '" + col.name + "'(" + std::to_string(col.length) + ") limit.");
                                   }
                                   if (upper_col_type == "JSON" && !validateJsonFormat(arg))
                                   {
                                       throw std::runtime_error("column '" + col.name + "' is not a valid JSON string.");
                                   }
                                   if (upper_col_type == "DATE" && !validateDateTimeFormat(arg, "DATE"))
                                   {
                                       throw std::runtime_error("column '" + col.name + "' is not a valid DATE format (should be YYYY-MM-DD).");
                                   }
                                   if (upper_col_type == "TIME" && !validateDateTimeFormat(arg, "TIME"))
                                   {
                                       throw std::runtime_error("column '" + col.name + "' is not a valid TIME format (should be HH:MM:SS).");
                                   }
                                   if (upper_col_type == "DATETIME" && !validateDateTimeFormat(arg, "DATETIME"))
                                   {
                                       throw std::runtime_error("column '" + col.name + "' is not a valid DATETIME format (should be YYYY-MM-DD HH:MM:SS).");
                                   }
                               } 
                               // 浮点数相关检查
                               if constexpr (std::is_floating_point_v<T>)
                               {
                                   if (upper_col_type == "FLOAT" && (arg < -(std::numeric_limits<float>::max()) || arg > std::numeric_limits<float>::max()))
                                   {
                                       throw std::runtime_error("value out of FLOAT range for column '" + col.name + "'");
                                   }
                                   if (upper_col_type == "DOUBLE" && (arg < -(std::numeric_limits<double>::max()) || arg > std::numeric_limits<double>::max()))
                                   {
                                       throw std::runtime_error("value out of DOUBLE range for column '" + col.name + "'");
                                   }
                               } },
                           current_val.value());
                if (col.auto_increment && value_provided)
                {
                    int64_t count;
                    if (col.type == "TINYINT")
                    {
                        count = static_cast<int64_t>(std::get<int8_t>(current_val.value()));
                    }
                    else if (col.type == "SMALLINT")
                    {
                        count = static_cast<int64_t>(std::get<int16_t>(current_val.value()));
                    }
                    else if (col.type == "INT")
                    {
                        count = static_cast<int64_t>(std::get<int32_t>(current_val.value()));
                    }
                    else if (col.type == "BIGINT")
                    {
                        count = std::get<int64_t>(current_val.value());
                    }
                    else if (col.type == "TINYINT UNSIGNED")
                    {
                        count = static_cast<int64_t>(std::get<uint8_t>(current_val.value()));
                    }
                    else if (col.type == "SMALLINT UNSIGNED")
                    {
                        count = static_cast<int64_t>(std::get<uint16_t>(current_val.value()));
                    }
                    else if (col.type == "INT UNSIGNED")
                    {
                        count = static_cast<int64_t>(std::get<uint32_t>(current_val.value()));
                    }
                    else if (col.type == "BIGINT UNSIGNED")
                    {
                        count = static_cast<int64_t>(std::get<uint64_t>(current_val.value()));
                    }
                    else
                    {
                        throw std::runtime_error("Unknown type: " + col.type);
                    }
                    if (count > table_schema.auto_increment_counters[col.name])
                    {
                        catalog_.setAutoIncrementCounter(table_name, col.name, count + 1);
                    }
                }
                // 检查唯一约束
                if ((col.primary || col.unique) && catalog_.uniqueConstraintExists(table_name, col.name, current_val))
                {
                    throw std::runtime_error("value for'" + col.name + "' violates the UNIQUE constraint.");
                }
            }
        }

        // --- 阶段 3: 所有检查通过后,为新的唯一值注册约束 ---
        for (size_t i = 0; i < table_schema.cols.size(); ++i)
        {
            if ((table_schema.cols[i].primary || table_schema.cols[i].unique) && row[i].has_value())
            {
                catalog_.createUniqueConstraint(table_name, table_schema.cols[i].name, row[i]);
            }
        }

        return {true, ""};
    }
    catch (std::exception &e)
    {
        logger_.Error(e.what());
        return {false, e.what()};
    }
}

bool Executor::executeCreateIndex(const std::string &table_name,
                                  const std::string &index_name,
                                  const std::vector<std::string> &columns,
                                  bool is_unique)
{
    try
    {
        // 1. 检查表是否存在
        if (!catalog_.tableExists(table_name))
        {
            logger_.Error("Table '" + table_name + "' does not exist.");
            throw std::runtime_error("Table '" + table_name + "' does not exist.");
        }

        // 2. 检查列是否存在
        auto table_schema = catalog_.getTableSchema(table_name);
        for (const auto &col : columns)
        {
            bool found = false;
            for (const auto &table_col : table_schema.cols)
            {
                if (table_col.name == col)
                {
                    found = true;
                    break;
                }
            }
            if (!found)
            {
                logger_.Error("Column '" + col + "' does not exist in table '" + table_name + "'");
                throw std::runtime_error("Column '" + col + "' does not exist in table '" + table_name + "'");
            }
        }
        if (columns.size() == 0)
        {
            logger_.Error("Index must have at least one column.");
            throw std::runtime_error("Index must have at least one column.");
        }
        else if (columns.size() > 1)
        {
            logger_.Error("Multi-column index is not supported.");
            throw std::runtime_error("Multi-column index is not supported.");
        }
        // 2. 在目录中创建索引元数据(先创建元数据,获得root_page_id)
        if (!catalog_.createIndex(table_name, index_name, columns, 0, is_unique))
        {
            throw std::runtime_error("Create index metadata failed.");
        }
        // 3. 构建新的B+树
        BPlusTree new_index_tree(table_name, index_name, catalog_);
        if (!new_index_tree.isEmpty())
        {
            throw std::runtime_error("New index tree is not empty.");
        }

        // 4. 获取表中的所有现有数据
        BPlusTree primary_tree(table_name, table_schema.primary_key_name, catalog_);
        auto res = primary_tree.ScanAll();
        std::vector<Row> all_rows = std::get<std::vector<Row>>(res);
        // 5. 遍历每一行,将数据填充到新的二级索引中
        for (const auto &row : all_rows)
        {
            BPlusTreeKey primary_key = extractPrimaryKeyFromRow(row, table_schema);
            BPlusTreeKey secondary_key = extractKeyFromRow(row, catalog_.getIndexSchema(table_name, index_name), table_schema);
            new_index_tree.Insert(secondary_key, primary_key, catalog_);
        }

        logger_.Info("在表 " + table_name + " 上成功创建并填充索引 " + index_name);
        return true;
    }
    catch (const std::exception &e)
    {
        logger_.Error("创建索引失败: " + std::string(e.what()));
        // 如果失败,应该回滚,删除已创建的索引元数据
        catalog_.dropIndex(table_name, index_name);
        throw e;
    }
}

bool Executor::executeDropIndex(const std::string &table_name,
                                const std::string &index_name)
{
    // 1. 检查索引是否存在
    if (!catalog_.indexExists(table_name, index_name))
    {
        logger_.Error("Index '" + index_name + "' does not exist in table '" + table_name + "'");
        throw std::runtime_error("Index '" + index_name + "' does not exist in table '" + table_name + "'");
    }
    // 2. 删除所有索引页
    BPlusTree index_tree(table_name, index_name, catalog_);
    index_tree.DeleteAll(catalog_);
    // 3. 调用目录管理器删除索引
    return catalog_.dropIndex(table_name, index_name);
}

size_t Executor::executeDelete(const std::string &table_name, const Condition &condition)
{
    try
    {
        uint32_t count = 0;
        auto table_schema = catalog_.getTableSchema(table_name);
        if (condition.isEmpty())
        {
            // 1. 删除所有行
            for (auto &index_schema : table_schema.indexes)
            {
                if (index_schema.index_name == table_schema.primary_key_name)
                    continue;
                BPlusTree secondary_tree(table_name, index_schema.index_name, catalog_);
                secondary_tree.DeleteAll(catalog_);
            }
            BPlusTree primary_tree(table_name, catalog_.getTableSchema(table_name).primary_key_name, catalog_);
            count += primary_tree.DeleteAll(catalog_);
            for (auto &col : table_schema.cols)
            {
                if (col.primary || col.unique)
                {
                    catalog_.deleteUniqueConstraint(table_name, col.name);
                }
            }
            return count;
        }
        // 1. 使用 Filter 逻辑找到所有需要删除的行
        auto rows_to_delete = executeFilter(condition, table_name);
        if (rows_to_delete.empty())
        {
            return 0; // 没有行符合条件,删除0行
        }

        // 2. 对每一行执行删除操作
        for (const auto &row : rows_to_delete)
        {
            // 2a. 提取主键
            BPlusTreeKey primary_key = extractPrimaryKeyFromRow(row, table_schema);

            // 2b. 从所有二级索引中删除条目
            for (const auto &index_schema : table_schema.indexes)
            {
                if (index_schema.index_name == table_schema.primary_key_name)
                    continue;

                BPlusTreeKey secondary_key = extractKeyFromRow(row, index_schema, table_schema);
                BPlusTree secondary_tree(table_name, index_schema.index_name, catalog_);
                secondary_tree.Delete(secondary_key, primary_key, catalog_);
            }

            // 2c. 最后从主键索引中删除整行数据
            BPlusTree primary_tree(table_name, table_schema.primary_key_name, catalog_);
            count += primary_tree.Delete(primary_key, catalog_);

            // 2d. 从目录中删除所有约束
            for (const auto &col : table_schema.cols)
            {
                if (col.primary || col.unique)
                {
                    catalog_.deleteUniqueConstraint(table_name, col.name, row[getColumnIndex(table_schema, col.name)]);
                }
            }
        }

        return count;
    }
    catch (const std::exception &e)
    {
        logger_.Error("delete failed: " + std::string(e.what()));
        throw e;
    }
}

size_t Executor::executeUpdate(const std::string &table_name, const Condition &condition, const std::unordered_map<std::string, Value> &update_values)
{
    try
    {
        auto table_schema = catalog_.getTableSchema(table_name);
        std::unordered_map<std::string, Value> new_values = update_values;
        for (const auto &[col_name, new_val] : update_values)
        {
            for (size_t i = 0; i < table_schema.cols.size(); ++i)
            {
                if (table_schema.cols[i].name == col_name)
                {
                    new_values[col_name] = convertValueToType(new_val, table_schema.cols[i].type);
                    break;
                }
            }
        }
        if (condition.isEmpty())
        {
            BPlusTree primary_tree(table_name, catalog_.getTableSchema(table_name).primary_key_name, catalog_);
            return primary_tree.UpdateAll(new_values, catalog_);
        }
        // 1. 找到所有需要更新的行
        auto old_rows = executeFilter(condition, table_name);
        if (old_rows.empty())
        {
            return 0;
        }

        // 2. 对每一行执行更新
        for (const auto &old_row : old_rows)
        {
            // 2a. 构造新行的数据
            Row new_row = old_row;
            BPlusTreeKey old_primary_key = extractPrimaryKeyFromRow(old_row, table_schema);
            BPlusTreeKey new_primary_key = old_primary_key;
            bool primary_key_changed = false;

            for (const auto &[col_name, new_val] : new_values)
            {
                for (size_t i = 0; i < table_schema.cols.size(); ++i)
                {
                    if (table_schema.cols[i].name == col_name)
                    {
                        new_row[i] = new_val;
                        if (table_schema.cols[i].primary)
                        {
                            primary_key_changed = true;
                            new_primary_key = new_val;
                        }
                        break;
                    }
                }
            }
            if (!primary_key_changed)
            {
                // 2b. 检查哪些索引键发生了变化
                for (const auto &index_schema : table_schema.indexes)
                {
                    if (index_schema.index_name == table_schema.primary_key_name)
                        continue;

                    BPlusTreeKey old_key = extractKeyFromRow(old_row, index_schema, table_schema);
                    BPlusTreeKey new_key = extractKeyFromRow(new_row, index_schema, table_schema);

                    if (old_key != new_key)
                    {
                        // 索引键变了,需要执行 "删除旧的,插入新的"
                        BPlusTree secondary_tree(table_name, index_schema.index_name, catalog_);
                        secondary_tree.Delete(old_key, old_primary_key, catalog_);
                        secondary_tree.Insert(new_key, old_primary_key, catalog_);
                    }
                }

                // 2c. 最后,更新主键索引中的行数据
                BPlusTree primary_tree(table_name, table_schema.primary_key_name, catalog_);
                primary_tree.Update(old_primary_key, new_values, catalog_);
            }
            else
            {
                for (const auto &index_schema : table_schema.indexes)
                {
                    if (index_schema.index_name == table_schema.primary_key_name)
                        continue;

                    BPlusTreeKey old_key = extractKeyFromRow(old_row, index_schema, table_schema);
                    BPlusTreeKey new_key = extractKeyFromRow(new_row, index_schema, table_schema);

                    // 主键变了,所有索引需要执行 "删除旧的,插入新的"
                    BPlusTree secondary_tree(table_name, index_schema.index_name, catalog_);
                    secondary_tree.Delete(old_key, old_primary_key, catalog_);
                    secondary_tree.Insert(new_key, new_primary_key, catalog_);
                }
                BPlusTree primary_tree(table_name, table_schema.primary_key_name, catalog_);
                primary_tree.Delete(old_primary_key, catalog_);
                primary_tree.Insert(new_primary_key, new_row, catalog_);
            }
        }

        return old_rows.size();
    }
    catch (const std::exception &e)
    {
        logger_.Error("更新失败: " + std::string(e.what()));
        throw e;
    }
}

bool Executor::executeDropTable(const std::string &table_name)
{
    try
    {
        auto schema = catalog_.getTableSchema(table_name);
        // 1. 删除所有索引
        for (const auto &index_schema : schema.indexes)
        {
            BPlusTree index_tree(table_name, index_schema.index_name, catalog_);
            index_tree.DeleteAll(catalog_);
        }
        // 2.删除表
        catalog_.dropTable(table_name);
        return true;
    }
    catch (const std::exception &e)
    {
        logger_.Error("删除表失败: " + std::string(e.what()));
        throw e;
    }
}

std::unordered_map<std::string, std::vector<ColumnSchema>> Executor::executeShowTables()
{
    std::unordered_map<std::string, std::vector<ColumnSchema>> result;
    auto schemas = catalog_.getAllSchemas();
    for (const auto &[table_name, schema] : schemas)
    {
        result[table_name] = schema.cols;
    }
    return result;
}

std::vector<IndexSchema> Executor::executeShowIndexes(const std::string &table_name)
{
    auto schema = catalog_.getTableSchema(table_name);
    return schema.indexes;
}

bool Executor::validateColumn(const ColumnSchema &col)
{
    std::string type = toUpperCase(col.type);
    if (col.name.empty())
    {
        logger_.Error("Column name is empty.");
        throw std::runtime_error("Column name is empty.");
    }
    if (type != "INT" && type != "VARCHAR" && type != "CHAR" && type != "TINYINT" && type != "DATE" && type != "TIME" && type != "DATETIME" && type != "SMALLINT" && type != "BIGINT" && type != "TEXT" && type != "TINYINT UNSIGNED" && type != "SMALLINT UNSIGNED" && type != "INT UNSIGNED" && type != "BIGINT UNSIGNED" && type != "JSON" && type != "BLOB" && type != "DOUBLE" && type != "FLOAT" && type != "BOOLEAN" && type != "BOOL")
    {
        logger_.Error("Column '" + col.name + "' has invalid type:" + type + ".");
        throw std::runtime_error("Column '" + col.name + "' has invalid type.");
    }
    if ((type == "VARCHAR" || type == "CHAR") && col.length == 0)
    {
        logger_.Error("Column '" + col.name + "' has invalid length.");
        throw std::runtime_error("Column '" + col.name + "' has invalid length.");
    }
    if (col.primary && col.nullable)
    {
        logger_.Error("Primary key column '" + col.name + "' cannot be nullable.");
        throw std::runtime_error("Primary key column '" + col.name + "' cannot be nullable.");
    }
    if (col.auto_increment && type != "INT" && type != "TINYINT" && type != "SMALLINT" && type != "BIGINT" && type != "TINYINT UNSIGNED" && type != "SMALLINT UNSIGNED" && type != "INT UNSIGNED" && type != "BIGINT UNSIGNED")
    {
        logger_.Error("Auto-increment column '" + col.name + "' must be of type INT.");
        throw std::runtime_error("Auto-increment column '" + col.name + "' must be of type INT.");
    }
    if (col.auto_increment && col.nullable)
    {
        logger_.Error("Auto-increment column '" + col.name + "' cannot be nullable.");
        throw std::runtime_error("Auto-increment column '" + col.name + "' cannot be nullable.");
    }
    if (col.unique && col.nullable)
    {
        logger_.Error("Unique column '" + col.name + "' cannot be nullable.");
        throw std::runtime_error("Unique column '" + col.name + "' cannot be nullable.");
    }
    if (col.default_value.has_value())
    {
        std::string default_value = col.default_value.value();
        if (type == "TINYINT")
        {
            auto v = std::stoll(default_value);
            if (v < std::numeric_limits<int8_t>::min() || v > std::numeric_limits<int8_t>::max())
            {
                logger_.Error("Default value '" + default_value + "' for column '" + col.name + "' is out of range.");
                throw std::runtime_error("Default value '" + default_value + "' for column '" + col.name + "' is out of range.");
            }
        }
        else if (type == "SMALLINT")
        {
            auto v = std::stoll(default_value);
            if (v < std::numeric_limits<int16_t>::min() || v > std::numeric_limits<int16_t>::max())
            {
                logger_.Error("Default value '" + default_value + "' for column '" + col.name + "' is out of range.");
                throw std::runtime_error("Default value '" + default_value + "' for column '" + col.name + "' is out of range.");
            }
        }
        else if (type == "INT")
        {
            auto v = std::stoll(default_value);
            if (v < std::numeric_limits<int32_t>::min() || v > std::numeric_limits<int32_t>::max())
            {
                logger_.Error("Default value '" + default_value + "' for column '" + col.name + "' is out of range.");
                throw std::runtime_error("Default value '" + default_value + "' for column '" + col.name + "' is out of range.");
            }
        }
        else if (type == "BIGINT")
        {
            auto v = std::stoll(default_value);
            if (v < std::numeric_limits<int64_t>::min() || v > std::numeric_limits<int64_t>::max())
            {
                logger_.Error("Default value '" + default_value + "' for column '" + col.name + "' is out of range.");
                throw std::runtime_error("Default value '" + default_value + "' for column '" + col.name + "' is out of range.");
            }
        }
        else if (type == "FLOAT")
        {
            auto v = std::stold(default_value);
            if (v < -(std::numeric_limits<float>::max()) || v > std::numeric_limits<float>::max())
            {
                logger_.Error("Default value '" + default_value + "' for column '" + col.name + "' is out of range.");
                throw std::runtime_error("Default value '" + default_value + "' for column '" + col.name + "' is out of range.");
            }
        }
        else if (type == "DOUBLE")
        {
            auto v = std::stold(default_value);
            if (v < -(std::numeric_limits<double>::max()) || v > std::numeric_limits<double>::max())
            {
                logger_.Error("Default value '" + default_value + "' for column '" + col.name + "' is out of range.");
                throw std::runtime_error("Default value '" + default_value + "' for column '" + col.name + "' is out of range.");
            }
        }
        else if (type == "VARCHAR" || type == "CHAR")
        {
            if (default_value.size() > col.length)
            {
                logger_.Error("Default value '" + default_value + "' for column '" + col.name + "' is too long.");
                throw std::runtime_error("Default value '" + default_value + "' for column '" + col.name + "' is too long.");
            }
        }
        else if (type == "TINYINT UNSIGNED")
        {
            auto v = std::stoll(default_value);
            if (v < std::numeric_limits<uint8_t>::min() || v > std::numeric_limits<uint8_t>::max())
            {
                logger_.Error("Default value '" + default_value + "' for column '" + col.name + "' is out of range.");
                throw std::runtime_error("Default value '" + default_value + "' for column '" + col.name + "' is out of range.");
            }
        }
        else if (type == "SMALLINT UNSIGNED")
        {
            auto v = std::stoll(default_value);
            if (v < std::numeric_limits<uint16_t>::min() || v > std::numeric_limits<uint16_t>::max())
            {
                logger_.Error("Default value '" + default_value + "' for column '" + col.name + "' is out of range.");
                throw std::runtime_error("Default value '" + default_value + "' for column '" + col.name + "' is out of range.");
            }
        }
        else if (type == "INT UNSIGNED")
        {
            auto v = std::stoll(default_value);
            if (v < std::numeric_limits<uint32_t>::min() || v > std::numeric_limits<uint32_t>::max())
            {
                logger_.Error("Default value '" + default_value + "' for column '" + col.name + "' is out of range.");
                throw std::runtime_error("Default value '" + default_value + "' for column '" + col.name + "' is out of range.");
            }
        }
        else if (type == "BIGINT UNSIGNED")
        {
            auto v = std::stoll(default_value);
            if (v < std::numeric_limits<uint64_t>::min() || v > std::numeric_limits<uint64_t>::max())
            {
                logger_.Error("Default value '" + default_value + "' for column '" + col.name + "' is out of range.");
                throw std::runtime_error("Default value '" + default_value + "' for column '" + col.name + "' is out of range.");
            }
        }
        else if (type == "DATE")
        {
            if (!validateDateTimeFormat(default_value, "DATE"))
            {
                logger_.Error("Default value '" + default_value + "' for column '" + col.name + "' is not a valid date.");
                throw std::runtime_error("Default value '" + default_value + "' for column '" + col.name + "' is not a valid date.");
            }
        }
        else if (type == "TIME")
        {
            if (!validateDateTimeFormat(default_value, "TIME"))
            {
                logger_.Error("Default value '" + default_value + "' for column '" + col.name + "' is not a valid time.");
                throw std::runtime_error("Default value '" + default_value + "' for column '" + col.name + "' is not a valid time.");
            }
        }
        else if (type == "DATETIME")
        {
            if (!validateDateTimeFormat(default_value, "DATETIME"))
            {
                logger_.Error("Default value '" + default_value + "' for column '" + col.name + "' is not a valid datetime.");
                throw std::runtime_error("Default value '" + default_value + "' for column '" + col.name + "' is not a valid datetime.");
            }
        }
        else if (type == "JSON")
        {
            if (!validateJsonFormat(default_value))
            {
                logger_.Error("Default value '" + default_value + "' for column '" + col.name + "' is not a valid json.");
                throw std::runtime_error("Default value '" + default_value + "' for column '" + col.name + "' is not a valid json.");
            }
        }
        else if (type == "BOOL" || type == "BOOLEAN")
        {
            std::string dv = toUpperCase(default_value);
            if (dv != "TRUE" && dv != "FALSE" && dv != "0" && dv != "1")
            {
                logger_.Error("Default value '" + default_value + "' for column '" + col.name + "' is not a valid boolean.");
                throw std::runtime_error("Default value '" + default_value + "' for column '" + col.name + "' is not a valid boolean.");
            }
        }
    }
    return true;
}

bool Executor::executeAlterTable(const std::string &table_name, std::optional<std::string> rename_table_name, std::optional<std::string> modified_comment,
                                 std::optional<std::unordered_map<std::string, ModifiedColumn>> modified_cols, std::optional<std::unordered_map<std::string, std::string>> modified_indexes, std::optional<std::vector<std::string>> drop_cols, std::optional<std::vector<ColumnSchema>> add_cols)
{
    // --- 阶段 1: 处理元数据和不需要重写数据的操作 ---
    if (rename_table_name.has_value())
    {
        catalog_.renameTable(table_name, rename_table_name.value());
        // 如果表被重命名,后续所有操作都应在新表名上进行
        return executeAlterTable(rename_table_name.value(), std::nullopt, modified_comment, modified_cols, modified_indexes, drop_cols, add_cols);
    }
    if (modified_comment.has_value())
    {
        catalog_.modifyComment(table_name, modified_comment.value());
    }
    if (modified_indexes.has_value())
    {
        for (auto &[old_name, new_name] : modified_indexes.value())
        {
            catalog_.renameIndex(table_name, old_name, new_name);
        }
    }

    // 检查是否需要对表数据进行重写
    if (!add_cols.has_value() && !drop_cols.has_value() && !modified_cols.has_value())
    {
        return true; // 如果只修改了表注释或索引名,到此结束
    }

    // --- 阶段 2: 读取现有数据并准备转换 ---
    TableSchema old_schema = catalog_.getTableSchema(table_name);
    BPlusTree primary_tree(table_name, old_schema.primary_key_name, catalog_);

    std::vector<Row> all_rows;
    bool has_data = !primary_tree.isEmpty();
    if (has_data)
    {
        auto result_variant = primary_tree.ScanAll();
        all_rows = std::get<std::vector<Row>>(result_variant);
    }

    TableSchema new_schema = old_schema;

    // --- 阶段 3: 在内存中模拟 Schema 变更并转换数据 ---

    // 3a. 处理新增列
    if (add_cols.has_value())
    {
        std::vector<Value> default_values_for_add;
        for (const auto &col_to_add : add_cols.value())
        {
            validateColumn(col_to_add);
            if (has_data)
            {
                if (!col_to_add.nullable && !col_to_add.default_value.has_value())
                {
                    throw std::runtime_error("The NOT NULL column '" + col_to_add.name + "'  cannot be added to the table with existing data, unless it has a default value.");
                }
                if (col_to_add.default_value.has_value())
                {
                    default_values_for_add.push_back(parseValueFromString(col_to_add.default_value.value(), col_to_add.type));
                }
                else
                {
                    default_values_for_add.push_back(std::nullopt); // 可空列的默认值是 NULL
                }
            }
        }
        catalog_.addColumns(table_name, add_cols.value());
        if (has_data)
        {
            for (auto &row : all_rows)
            {
                for (const auto &val : default_values_for_add)
                {
                    row.push_back(val);
                }
            }
        }
    }

    // 3b. 处理删除列
    if (drop_cols.has_value())
    {
        std::vector<size_t> indices_to_drop;
        for (const auto &col_name_to_drop : drop_cols.value())
        {
            size_t index = getColumnIndex(new_schema, col_name_to_drop);
            if (new_schema.cols[index].primary)
            {
                throw std::runtime_error("Cannot drop the primary key column'" + col_name_to_drop + "'.");
            }
            indices_to_drop.push_back(index);
        }
        std::sort(indices_to_drop.rbegin(), indices_to_drop.rend()); // 从后往前删,避免索引失效
        catalog_.dropColumns(table_name, drop_cols.value());
        if (has_data)
        {
            for (auto &row : all_rows)
            {
                for (size_t index : indices_to_drop)
                {
                    row.erase(row.begin() + index);
                }
            }
        }
    }

    // 3c. 处理修改列
    if (modified_cols.has_value())
    {
        std::unordered_map<std::string, int> need_to_auto_increment;
        std::unordered_map<std::string, std::set<Value>> need_to_unique;
        std::vector<std::string> need_to_drop_auto_increment;
        std::vector<std::string> need_to_drop_unique;
        for (const auto &[old_name, mod_info] : modified_cols.value())
        {
            size_t index = getColumnIndex(new_schema, old_name);
            ColumnSchema &col_to_modify = new_schema.cols[index];

            // --- 数据验证阶段:在修改任何东西之前,检查所有约束 ---
            if (has_data)
            {
                // 验证1: 尝试将列改为 NOT NULL
                if (mod_info.nullable.has_value() && !mod_info.nullable.value() && col_to_modify.nullable)
                {
                    for (const auto &row : all_rows)
                    {
                        if (!row[index].has_value())
                        {
                            throw std::runtime_error("cannot modify column '" + old_name + "' to NOT NULL, because it contains NULL values.");
                        }
                    }
                }

                // 验证2: 尝试添加 UNIQUE 约束
                if (mod_info.unique.has_value() && mod_info.unique.value() && !col_to_modify.unique)
                {
                    std::set<Value> existing_values;
                    for (const auto &row : all_rows)
                    {
                        if (row[index].has_value())
                        { // UNIQUE 约束不作用于 NULL 值
                            if (!existing_values.insert(row[index]).second)
                            {
                                throw std::runtime_error("cannot add UNIQUE constraint to'" + old_name + "', because it contains duplicate values.");
                            }
                        }
                    }
                    // 验证通过,准备好要传递给 Catalog 的数据集
                    need_to_unique[col_to_modify.name] = existing_values;
                }
                if (mod_info.auto_increment.has_value() && mod_info.auto_increment.value() && !col_to_modify.auto_increment)
                {
                    if (toUpperCase(col_to_modify.type) != "INT" && toUpperCase(col_to_modify.type) != "BIGINT" && toUpperCase(col_to_modify.type) != "SMALLINT" && toUpperCase(col_to_modify.type) != "TINYINT" && toUpperCase(col_to_modify.type) != "INT UNSIGNED" && toUpperCase(col_to_modify.type) != "BIGINT UNSIGNED" && toUpperCase(col_to_modify.type) != "SMALLINT UNSIGNED" && toUpperCase(col_to_modify.type) != "TINYINT UNSIGNED")
                    {
                        throw std::runtime_error("AUTO_INCREMENT can only be added to INT, BIGINT, SMALLINT, TINYINT, INT UNSIGNED, BIGINT UNSIGNED, SMALLINT UNSIGNED, TINYINT UNSIGNED columns.");
                    }
                    int64_t max_val = 0;
                    for (const auto &row : all_rows)
                    {
                        if (row[index].has_value())
                        {
                            int64_t current_val = std::visit([](auto &&arg) -> int64_t
                                                             {
                                using T = std::decay_t<decltype(arg)>;
                                if constexpr (std::is_same_v<T, std::string>) {
                                    // 如果是字符串,则尝试用 stoll 解析
                                    return std::stoll(arg);
                                } else if constexpr (std::is_arithmetic_v<T>) {
                                    // 如果是任何数字或bool类型,则进行 static_cast
                                    return static_cast<int64_t>(arg);
                                }
                                // 对于其他类型,抛出异常
                                throw std::runtime_error("Cannot convert value to int64_t."); }, row[index].value());
                            if (current_val > max_val)
                            {
                                max_val = current_val;
                            }
                        }
                    }
                    // 准备好要传递给 Catalog 的最大值
                    need_to_auto_increment[col_to_modify.name] = static_cast<int>(max_val);
                }

                // 验证3: 尝试缩短 VARCHAR 长度
                if (toUpperCase(col_to_modify.type) == "VARCHAR" && mod_info.length.has_value() && mod_info.length.value() < col_to_modify.length)
                {
                    for (const auto &row : all_rows)
                    {
                        if (row[index].has_value())
                        {
                            const std::string &s = std::get<std::string>(row[index].value());
                            if (s.length() > mod_info.length.value())
                            {
                                throw std::runtime_error("cannot modify hte length of column'" + old_name + "' to " + std::to_string(mod_info.length.value()) + ", because it contains values longer than " + std::to_string(mod_info.length.value()) + " characters.");
                            }
                        }
                    }
                }

                // 验证4: 尝试进行数据类型转换 (预检查)
                if (mod_info.type.has_value() && toUpperCase(mod_info.type.value()) != toUpperCase(col_to_modify.type))
                {
                    for (const auto &row : all_rows)
                    {
                        if (row[index].has_value())
                        {
                            try
                            {
                                // 使用新的、带范围检查的转换函数进行模拟转换
                                convertValueToType(row[index], mod_info.type.value());
                            }
                            catch (const std::exception &e)
                            {
                                throw std::runtime_error("无法将列 '" + old_name + "' 的现有值转换为新类型 '" + mod_info.type.value() + "': " + e.what());
                            }
                        }
                    }
                }
            }

            // --- 元数据和数据转换阶段 ---

            // 检查对主键的非法修改
            if (col_to_modify.primary)
            {
                if (mod_info.primary.has_value() && !mod_info.primary.value())
                {
                    throw std::runtime_error("cannot modify the primary key column '" + old_name + "' to NOT PRIMARY KEY.");
                }
                if (mod_info.nullable.has_value() && mod_info.nullable.value())
                {
                    throw std::runtime_error("The primary key column '" + old_name + "' must be NOT NULL.");
                }
            }
            else
            {
                if (mod_info.primary.has_value() && mod_info.primary.value())
                {
                    throw std::runtime_error("cannot modify the column '" + old_name + "' to PRIMARY KEY, because primary key is existed.");
                }
            }

            // 执行数据类型转换
            if (has_data && mod_info.type.has_value() && toUpperCase(mod_info.type.value()) != toUpperCase(col_to_modify.type))
            {
                for (auto &row : all_rows)
                {
                    // 直接使用新的转换函数，它会处理所有情况并抛出异常
                    row[index] = convertValueToType(row[index], mod_info.type.value());
                }
            }
            if (mod_info.unique.has_value() && !mod_info.unique.value())
            {
                need_to_drop_unique.push_back(col_to_modify.name);
            }
            if (mod_info.auto_increment.has_value() && !mod_info.auto_increment.value())
            {
                need_to_drop_auto_increment.push_back(col_to_modify.name);
            }
        }
        auto mod_cols_non_const = modified_cols.value();
        catalog_.modifyColumns(table_name, mod_cols_non_const, need_to_auto_increment, need_to_unique, need_to_drop_auto_increment, need_to_drop_unique);
    }

    // --- 阶段 4: 清空旧表并用新数据和新 Schema 重建 ---
    if (has_data)
    {
        primary_tree.DeleteAll(catalog_);
    }

    // 然后再用更新后的 new_schema (现在应与 catalog 中的一致) 重建表
    new_schema = catalog_.getTableSchema(table_name); // 从 catalog 重新获取最终的 schema
    // 如果有数据,则重新插入
    if (has_data && !all_rows.empty())
    {
        // 重新获取 B+ 树实例,因为它可能依赖于更新后的 Schema
        BPlusTree new_primary_tree(table_name, new_schema.primary_key_name, catalog_);

        // 重新构建所有索引
        std::vector<std::unique_ptr<BPlusTree>> secondary_trees;
        for (const auto &index_schema : new_schema.indexes)
        {
            if (index_schema.index_name != new_schema.primary_key_name)
            {
                auto secondary_tree = std::make_unique<BPlusTree>(table_name, index_schema.index_name, catalog_);
                secondary_tree->DeleteAll(catalog_); // 清空旧的二级索引
                secondary_trees.push_back(std::move(secondary_tree));
            }
        }

        // 重新获取二级索引的 schema
        std::vector<IndexSchema> secondary_index_schemas;
        for (const auto &index_schema : new_schema.indexes)
        {
            if (index_schema.index_name != new_schema.primary_key_name)
            {
                secondary_index_schemas.push_back(index_schema);
            }
        }

        for (const auto &row : all_rows)
        {
            BPlusTreeKey pk = extractPrimaryKeyFromRow(row, new_schema);
            new_primary_tree.Insert(pk, row, catalog_);

            // 重新填充二级索引
            for (size_t i = 0; i < secondary_trees.size(); ++i)
            {
                BPlusTreeKey sk = extractKeyFromRow(row, secondary_index_schemas[i], new_schema);
                secondary_trees[i]->Insert(sk, pk, catalog_);
            }
        }
    }

    return true;
}